package com.dec.kks.etl.producer;

import org.apache.commons.lang3.RandomUtils;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.List;
import java.util.Map;

public class WordProducerMain {

    /**
     * <p/>
     * <li>Description:KKSDATAWriteToKafka</li>
     * <li>@author: liukailong <kailong.liu@cdcalabar.com> </li>
     * <li>Date: 18-4-20 上午10:16</li>
     * <li>@version: 2.0.0 </li>
     * <li>@since JDK 1.8 </li>
     */

    private static final Logger LOGGER = LoggerFactory.getLogger(WordProducerMain.class);



    public static void main(String[] args) throws Exception {
        String topic = "spark-topic-streaming";
        String path = "/home/hdfs/soft/IdeaProjects/dec-kks-etl/src/main/resources";
        int sleepTime = 1000;

        List<String> chars = Arrays.asList("a","b","c","d","e","d","f");
        int size = chars.size();
        Map config = DECConfigUtil.loadConfig(path,1);
        KafkaProducer producer = DECKafkaUtil.initProducer(config);
        while (true) {
            Integer msgId = Math.toIntExact(System.currentTimeMillis() / 1000);
            int idx = RandomUtils.nextInt(0,size);
            String msg = chars.get(idx);
            DECKafkaUtil.sendMsg(producer, topic,msgId+"", msg);
            if (msgId == msgId + 300 * 10000) {
                break;
            }
            System.out.println(msgId+":"+msg);
            Thread.sleep(sleepTime);
        }
    }



}
